package demo1;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @Author: wzg
 * @Date: 2021/4/14 下午4:20
 */
public class Producer {
    public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        Map<String,Object> configs = new HashMap<>();
        configs.put("bootstrap.servers", "192.168.50.117:9092");
        configs.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        configs.put("acks", "all");
        configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my_transactional_id");
        KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(configs);
        producer.initTransactions();
        producer.beginTransaction();
        // 用于封装Producer的消息
        for (int i = 1; i <= 2; i++) {
            ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>( "topic_1",
                    // 主题名称
                    0,
                    // 分区编号，现在只有一个分区，所以是0
                    i, // 数字作为key
                    "message "+i // 字符串作为value
            );
            // 发送消息，同步等待消息的确认
            producer.send(record).get(3_000, TimeUnit.MILLISECONDS);
        }
        producer.abortTransaction();
        // 关闭生产者
        producer.close();
    }
}
